package com.charles.bigdata.mapreduce.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCountMain {

	public static void main(String[] args) throws Exception {
		// 1.创建一个job和任务入口
		Job job = Job.getInstance(new Configuration());
		job.setJarByClass(WordCountMain.class); // main方法所在的class

		// 2.指定job的mapper和输出的类型<k2 v2>
		job.setMapperClass(WordCountMapper.class);// 指定Mapper类
		job.setMapOutputKeyClass(Text.class); // k2的类型
		job.setMapOutputValueClass(IntWritable.class); // v2的类型

		// 3.指定job的reducer和输出的类型<k4  v4>
		job.setReducerClass(WordCountReducer.class);// 指定Reducer类
		job.setOutputKeyClass(Text.class); // k4的类型
		job.setOutputValueClass(IntWritable.class); // v4的类型

		// 4.指定job的输入和输出
		// windows测试用的
		FileInputFormat.setInputPaths(job, new Path("E:\\temp\\word.txt")); // 待统计的数据问题
		FileOutputFormat.setOutputPath(job, new Path("E:\\temp\\temp")); // 输出路径

		//		// 生产环境用的
		//		//多路径判断
		//		if (args.length < 2) {
		//			System.out.println("参数数量不对，至少两个以上参数：<数据文件输出路径>、<输入路径...>");
		//			System.exit(1);
		//		}
		//		// 输出结果路径
		//		String dataOutput = args[0];
		//		// 多个路输入径
		//		String[] inputs = new String[args.length - 1];
		//		System.arraycopy(args, 1, inputs, 0, inputs.length);
		//
		//		//将输出路径和输入路径放入Path中
		//		Path[] inputPathes = new Path[inputs.length];
		//		for (int i = 0; i < inputs.length; i++) {
		//			inputPathes[i] = new Path(inputs[i]);
		//		}
		//		Path outputPath = new Path(dataOutput);
		//		FileInputFormat.setInputPaths(job, inputPathes); // 待统计的数据问题
		//		FileOutputFormat.setOutputPath(job, outputPath); // 输出路径

		// 5.执行job
		job.waitForCompletion(true);
	}
}